#include "Spark/Net/Impl/Connector.h"
#include "Spark/Net/Channel.h"
#include "Spark/Utils/Logger.h"
#include "Spark/Net/EventLoop.h"
#include <errno.h>
#include <unistd.h>

static const char* stateString(Connector::ConnectorState state)
{
    switch (state)
    {
        case Connector::ConnectorState::Connected:
            return "Connected";
        case Connector::ConnectorState::Connecting:
            return "Connecting";
        case Connector::ConnectorState::Disconnected:
            return "Disconnected";
    }
    return "";
}

Connector::Connector(EventLoop* loop, SocketAddr addr)
    : loop_(loop),
      peeraddr_(std::move(addr)),
      state_{ConnectorState::Disconnected}

{
    LOG_TRACE << "Connector[" << this << "]";
}

Connector::~Connector()
{
    LOG_TRACE << "Connector[" << this << "]";
    assert(!channel_);
}

void Connector::start()
{
    connect_.store(true);
    loop_->runInLoop([this] { startInLoop(); });
}

void Connector::restart()
{
    state_ = ConnectorState::Disconnected;
    retryInterval_ = kInitRetryDelayMs;
    connect_.store(true);
    startInLoop();
}

void Connector::stop()
{
    connect_.store(false);
    loop_->queueInLoop([this] { this->stopInLoop(); });
}

void Connector::startInLoop()
{
    assert(state_ == ConnectorState::Disconnected);
    if (connect_)
    {
        connect();
    }
    else
    {
        LOG_DEBUG << "connect_ = false";
    }
}

void Connector::stopInLoop()
{
    if (state_ == ConnectorState::Connecting)
    {
        state_ = ConnectorState::Disconnected;
        int sockfd = removeAndResetChannel();
        retry(sockfd);
    }
}

void Connector::connect()
{
    int sockfd = Socket::createSocket(peeraddr_.family());
    int ret = Socket::connect(sockfd, &peeraddr_);
    int savedErrno = errno;
    switch (savedErrno)
    {
        case 0:
        case EINPROGRESS:
        case EINTR:
        case EISCONN:
            LOG_TRACE << "connecting to fd[" << sockfd << "]";
            connecting(sockfd);
            break;

        case EAGAIN:
        case EADDRINUSE:
        case EADDRNOTAVAIL:
        case ECONNREFUSED:
        case ENETUNREACH:
            retry(sockfd);
            break;

        case EACCES:
        case EPERM:
        case EAFNOSUPPORT:
        case EALREADY:
        case EBADF:
        case EFAULT:
        case ENOTSOCK:
            LOG_SYSERR << "connect error in Connector::startInLoop "
                       << savedErrno;
            ::close(sockfd);
            if (errorCallback_)
                errorCallback_();
            break;

        default:
            LOG_SYSERR << "Unexpected error in Connector::startInLoop "
                       << savedErrno;
            ::close(sockfd);
            if (errorCallback_)
                errorCallback_();
            break;
    }
}

void Connector::connecting(int sockfd)
{
    state_ = ConnectorState::Connecting;
    assert(!channel_);
    channel_.reset(new Channel(loop_, sockfd));
    channel_->setCallback(CBType::WRITE, [this] { this->handleWrite(); });
    channel_->setCallback(CBType::CLOSE, [this] { this->handleError(); });
    channel_->setCallback(CBType::ERROR, [this] { this->handleError(); });
    LOG_TRACE << "connecting:" << sockfd;
    channel_->addWriteEvent();
}

int Connector::removeAndResetChannel()
{
    channel_->disableAll();
    channel_->remove();
    int sockfd = channel_->fd();
    LOG_TRACE << "Connector::removeAndResetChannel, fd[" << sockfd << "]";
    loop_->queueInLoop([this] { this->channel_.reset(); });
    return sockfd;
}

void Connector::handleWrite()
{
    LOG_TRACE << "Connector::handleWrite" << stateString(state_);
    if (state_ == ConnectorState::Connecting)
    {
        int sockfd = removeAndResetChannel();
        int err = Socket::getSockErrno(sockfd);
        if (err)
        {
            LOG_WARN << "Connector::handleWrite - SO_ERROR: "
                     << strerror_tl(err);
            retry(sockfd);
            if (errorCallback_)
                errorCallback_();
        }
        else if (Socket::isSelfConnected(sockfd))
        {
            LOG_WARN << "Connector::writeCallbcak - SelfConnected";
            retry(sockfd);
            if (errorCallback_)
                errorCallback_();
        }
        else
        {
            state_ = ConnectorState::Connected;
            if (connect_)
                newConnectionCallback_(sockfd);
            else
                ::close(sockfd);
        }
    }
    else
    {
        assert(state_ == ConnectorState::Disconnected);
    }
}

void Connector::handleError()
{
    LOG_ERROR << "Connector::handleError state=" << stateString(state_);
    if (state_ == ConnectorState::Connecting)
    {
        state_ = ConnectorState::Disconnected;
        int sockfd = removeAndResetChannel();
        int err = Socket::getSockErrno(sockfd);
        LOG_TRACE << "SO_ERROR: " << err << " " << strerror_tl(err);
        retry(sockfd);
        if (errorCallback_)
            errorCallback_();
    }
}

void Connector::retry(int sockfd)
{
    LOG_TRACE << "Connector::retry, state=" << stateString(state_)
              << " retry=" << retry_;
    ::close(sockfd);
    state_ = ConnectorState::Disconnected;
    if (!retry_)
        return;
    if (connect_)
    {
        auto thisPtr = shared_from_this();
        LOG_INFO << "Connector::retry connecting to "
                 << peeraddr_.toStringPort() << " in " << retryInterval_
                 << " milliseconds";
        loop_->runAfter(std::chrono::milliseconds(retryInterval_),
                        [thisPtr] { thisPtr->startInLoop(); });
        retryInterval_ *= 2;
        if (retryInterval_ > kMaxRetryDeleyMs)
            retryInterval_ = kMaxRetryDeleyMs;
    }
    else
    {
        LOG_DEBUG << "connect_ = false";
    }
}
